#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/helper.hpp"
#include "../mqcommon/log.hpp"
#include "../mqcommon/mq_meg.pb.h"
#include "../mqcommon/mq_proto.pb.h"

#include "mq_consumer.hpp"

#include <iostream>
#include <mutex>
#include <unordered_map>
#include <condition_variable>

namespace pprmq
{
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codec)
            :_cid(UUIDHelper::uuid()),_conn(conn),_codec(codec)
        {}
        ~Channel() { basicCancel(); }
        std::string cid() { return _cid; }
        bool openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            // 构建请求消息
            OpenChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            // 通过_codec将请求消息req序列化，再通过_conn发送到服务器
            _codec->send(_conn,req);
            // 阻塞当前线程，直到收到服务器返回的、与 rid 匹配的响应
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            // 构建请求消息
            CloseChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        bool declareExchange(
            const std::string& name, 
            ExchangeType type,
            const bool durable,
            const bool auto_delete,
            google::protobuf::Map<std::string, std::string>& args)
        {
            std::string rid = UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void deleteExchange(const std::string &name)
        {
            std::string rid = UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_exchange_name(name);

            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        bool declareQueue(
            const std::string& qname, 
            bool durable, 
            bool exclusive, 
            bool auto_delete,
            google::protobuf::Map<std::string, std::string>& args)
        {
            std::string rid = UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_queue_name(qname);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);

            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
                
        void deleteQueue(const std::string &qname)
        {
            std::string rid = UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_queue_name(qname);

            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        bool queueBind(
            const std::string& ename,
            const std::string& qname,
            const std::string& binding_key)
        {
            std::string rid = UUIDHelper::uuid();
            queueBindRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(binding_key);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }

        void queueUnBind(const std::string& ename,const std::string& qname)
        {
            std::string rid = UUIDHelper::uuid();
            queueUnBindRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);

            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        void basicPublish(
            const std::string &ename, 
            const BasicProperties *bp, 
            const std::string &body)
        {
            std::string rid = UUIDHelper::uuid();
            basicPublishRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_exchange_name(ename);
            req.set_body(body);

            if(bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery(bp->delivery());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }

            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        void basicAck(const std::string &msgid)
        {
            std::string rid = UUIDHelper::uuid();
            basicAckRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msgid);

            _codec->send(_conn,req);
            waitResponse(rid);
            return ;
        }

        bool basicConsume(
            const std::string &consumer_tag,
            const std::string &queue_name,
            bool auto_ack,
            const ConsumerCallback &cb)
        {
            // 检查当前通道是否已订阅队列（避免重复订阅）
            if(_consumer.get() != nullptr) 
            {
                DLOG("当前信道已订阅其他队列消息");
                return false; 
            }
            // 构建订阅请求消息
            std::string rid = UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_queue_name(queue_name);
            req.set_consumer_tag(consumer_tag);
            req.set_auto_ack(auto_ack);
            // 发送请求并同步等待服务器确认
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = waitResponse(rid);
            // 验证服务器响应并创建本地消费者
            if(resp->ok() == false)
            {
                DLOG("添加订阅失败");
                return false;
            }
            _consumer = std::make_shared<Consumer>(consumer_tag,queue_name,auto_ack,cb);
            DLOG("添加订阅成功");
            return true;
        }

        void basicCancel()
        {
            // 若当前通道没有关联的消费者，直接返回，避免无效操作
            if (_consumer.get() == nullptr) { return ; }
            
            std::string rid = UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            req.set_consumer_tag(_consumer->_tag);
            req.set_queue_name(_consumer->_qname);

            _codec->send(_conn,req);
            waitResponse(rid);

            _consumer.reset(); // 重置智能指针，释放消费者对象

            return ;
        }
    public:
        // 连接收到基础响应后，向hash_map中添加响应
        void putBasicResponse(const basicCommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(),resp));
            _cv.notify_all();
        }

        // 连接收到消息推送后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理
        void consume(const basicConsumeResponsePtr& resp)
        {
            if(_consumer.get() == nullptr)
            {
                DLOG("消息处理时,未找到订阅者信息");
                return;
            }
            if(_consumer->_tag != resp->consumer_tag())
            {
                DLOG("收到的推送消息中的消费者标识,与当前信道消费者标识不一致");
                return ;
            }
                                // 消费者标签，属性，主体
            _consumer->_callback(resp->consumer_tag(),resp->mutable_properties(),resp->body());
        }
    private:
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // while(condition()) _cv.wait();
            _cv.wait(lock,[&rid,this](){
                return _basic_resp.find(rid) != _basic_resp.end();
            });
            basicCommonResponsePtr basic_resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }
    private:
        std::string _cid;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp; // 通过rid与响应的映射关系
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        Channel::ptr create(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel = std::make_shared<Channel>(conn,codec);
            _channels.insert(std::make_pair(channel->cid(),channel));
            return channel;
        }
        void remove(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr get(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string,Channel::ptr> _channels;
    };
}

#endif